package org.suian;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.channels.spi.AbstractSelector;
import java.nio.channels.spi.SelectorProvider;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Main {
    private static final int PORT = 1024;
    private static final int THREAD_POOL_SIZE = 10; // 线程池大小

    // 使用 ConcurrentHashMap 来跟踪活动的客户端连接
    private static final ConcurrentHashMap<SocketChannel, Boolean> activeConnections = new ConcurrentHashMap<>();
    private static final Lock lock = new ReentrantLock();
    public static void main(String[] args) throws IOException {
        Selector selector = null;
        ServerSocketChannel serverChannel = null;
        ExecutorService executorService = null;
        try {
            //打开一个选择器
            selector = SelectorProvider.provider().openSelector();

            //打开一个serverSocket通道
            serverChannel = ServerSocketChannel.open();
            serverChannel.configureBlocking(false);  //设置为非阻塞模式

            //绑定服务器地址和端口
            serverChannel.socket().bind(new InetSocketAddress(PORT));

            //将通道注册到选择器上，并监听连接请求
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);


            // 创建线程池
            executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

            while (true) {
                //阻塞等待就绪的通道，如果没有就绪的通道，则会一直阻塞在这里。
                selector.select();

                //获取已经准备好的选择键集合
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();

                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();//记得移除已处理的键，避免重复处理

                    try {
                        if (key.isAcceptable()) {
                            handleAccept(selector, executorService, key);
                        } else if (key.isReadable()) {
                            executorService.submit(() -> handleRead(key));
                        }
                    } catch (CancelledKeyException e) {
                        //处理已取消的键，避免程序崩溃
                        System.out.println("遇到已经取消的键："+e.getMessage());
                    }
                }
            }
        } catch (IOException e) {

            System.out.println("服务器异常：" + e.getMessage());
        } finally {
            // 清理资源
            closeSelector(selector);
            closeServerSocketChannel(serverChannel);
            shutdownExecutorService(executorService);

        }


    }

    private static void shutdownExecutorService(ExecutorService executorService) {
        if (executorService != null) {
            executorService.shutdown();
        }
    }

    private static void closeServerSocketChannel(ServerSocketChannel serverChannel) {
        if (serverChannel != null && serverChannel.isOpen()) {
            try {
                serverChannel.close();
            } catch (IOException e) {
                System.err.println("关闭服务器通道时发生异常: " + e.getMessage());
            }
        }
    }

    private static void closeSelector(Selector selector) {
        if (selector != null && selector.isOpen()) {
            try {
                selector.close();
            } catch (IOException e) {
                System.err.println("关闭选择器时发生异常: " + e.getMessage());
            }
        }
    }

    //处理读取客户端操作
    private static void handleRead(SelectionKey key) {
        SocketChannel clientChannel = (SocketChannel) key.channel();
        // 首先检查通道是否关闭
        if (!clientChannel.isOpen()) {
            return;
        }
        try {
            ByteBuffer buffer = ByteBuffer.allocate(256);
            int bytesRead ;
            lock.lock();
            try {
                // 检查通道是否关闭
                if (!clientChannel.isOpen()) {
                    return;
                }
                bytesRead = clientChannel.read(buffer);
            }  finally {
                lock.unlock();
            }

            if (bytesRead == -1) {
                lock.lock();
                try {
                    if (clientChannel.isOpen()) {
                        clientChannel.close();
                        // 从活动连接集合中移除已关闭的通道
                        activeConnections.remove(clientChannel);
                    }
                }finally {
                    lock.unlock();
                }
            } else if(bytesRead > 0) {
                buffer.flip();//调用flip()方法，buffer从写模式变为读模式
                byte[] bytes = new byte[buffer.remaining()];
                buffer.get(bytes);
                String message = new String(bytes);
                System.out.println("线程名"+Thread.currentThread().getName()+"接收到客户端消息：" + message);
                // 处理数据...
                // 写数据到客户端
                String replyMessage = "";
                switch (message) {
                    case "查询设备信息":
                        replyMessage = "服务器-->设备连接信息: " + clientChannel.getRemoteAddress();
                        System.out.println(replyMessage);
                        break;
                    case "查询设备数量":
                        replyMessage = "服务器-->当前连接设备总数: "+ activeConnections.size();
                        System.out.println(replyMessage);
                        break;
                }

                if (!replyMessage.equals("")) {
                    ByteBuffer echoBuffer = ByteBuffer.wrap(replyMessage.getBytes());
                    clientChannel.write(echoBuffer);
                }
            }
        } catch (IOException e) {
            String errorMessage = (e.getMessage() == null)? e.toString() : e.getMessage();
            System.err.println("处理读操作时发生异常: " + errorMessage);
            try {
                lock.lock();
                try {
                    if (clientChannel.isOpen()) {
                        clientChannel.close();
                        activeConnections.remove(clientChannel);
                    }
                } finally {
                    lock.unlock();
                }
            } catch (IOException ex) {
                System.err.println("关闭出错通道时发生异常: " + ex.getMessage());
            }
        }

    }

    //处理已接受的连接事件
    private static void handleAccept(Selector selector, ExecutorService executorService, SelectionKey key) {
        ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
        try {
            SocketChannel clientChannel = serverChannel.accept();
            clientChannel.configureBlocking(false);

            //将连接加入到活动连接集合中
            activeConnections.put(clientChannel, true);
            // 将客户端通道注册到选择器上，并监听读事件
            SelectionKey clientKey = clientChannel.register(selector, SelectionKey.OP_READ);

            // 可以在这里将新的连接任务提交给线程池处理，但简单的注册操作通常不需要
            // executorService.submit(() -> {
            //     // 处理新连接的任务
            // });

            // 打印设备的连接信息
            InetSocketAddress clientAddress = (InetSocketAddress) clientChannel.getRemoteAddress();
            System.out.println("设备连接: " + clientAddress.getAddress().getHostAddress() + ":" + clientAddress.getPort());

        } catch (IOException e) {
            System.err.println("处理新连接时发生异常: " + e.getMessage());
        }

    }
}